/**
 * Created by Administrator on 2017/11/1.
 * Copyright 2017-2018 汉威.智慧环保事业部 Limited.
 * All rights reserved.
 */
package com.hw.one.collect.radar.Listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hw.one.collect.radar.po.RadarSitePo;
import com.hw.one.collect.radar.po.SingleDataPo;
import com.hw.one.collect.radar.service.RadarService;
import com.hw.one.collect.util.Base64Util;
import com.hw.one.collect.util.ConvertUtil;
import com.hw.one.collect.util.MessageProducer;
import com.hw.one.collect.util.ZLibUtils;
import com.rabbitmq.client.Channel;
import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

public class SingleRadarReceiver implements ChannelAwareMessageListener {

    private static Logger log = Logger.getLogger(SingleRadarReceiver.class);
    private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Resource
    public RadarService radarService;

    @Resource
    private MessageProducer messageProducer;

    /**
     * Callback for processing a received Rabbit message.
     * <p>Implementors are supposed to process the given Message,
     * typically sending reply messages through the given Session.
     *
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        try {
            Calendar calendar = Calendar.getInstance();

            log.info("------------------------单条雷达数据接入------------------------");
            log.info("数据到的时间:" + sdf.format(new Date()));
            log.info("----------------------------------------------------------");

            String orgData = new String(message.getBody());
            SingleDataPo singleDataPo = null;
            // 如果收到数据为null，直接跳出；
            if (null != orgData) {
                singleDataPo = new SingleDataPo();
                // 根据行结束符进行分割
                String[] rowDatas = orgData.split("\\r\\n");
                if (null != rowDatas && rowDatas.length > 0) {
                    //解析第一行数据(包含设备编码，开始结束数据，经纬度，数据类型)
                    if (null != rowDatas[0]) {
                        String[] singleData = rowDatas[0].split(",");
                        if (null != singleData && singleData.length == 5) {
                            // 雷达编码
                            singleDataPo.setRADARCODE(singleData[0]);
                            RadarSitePo radarSitePo = radarService.getRadarSiteInfo(singleData[0]);
                            singleDataPo.setSITECODE(radarSitePo.getSiteCode());
                            singleDataPo.setSITENAME(radarSitePo.getSiteName());
                            singleDataPo.setSITETYPECODE(radarSitePo.getSiteTypeCode());
                            // 雷达中心点经纬度
                            singleDataPo.setMONITORTIME(singleData[1]);
                            String[] loc = singleData[2].split(" ");
                            if (null != loc && loc.length == 2) {
                                double[] locs = new double[2];
                                locs[0] = ConvertUtil.toDouble(loc[0]);
                                locs[1] = ConvertUtil.toDouble(loc[1]);
                                singleDataPo.setRADARLOC(locs);
                            }
                            // 该数据的经纬度
                            singleDataPo.setANGLE(ConvertUtil.toInteger(singleData[3]));
                            // 数据类型 C平面扫描；L定向扫描
                            singleDataPo.setDATATYPE(singleData[4]);

                            int dataLen = 1000;
                            double[] distance = new double[dataLen + 1];
                            double[] pm2_5 = new double[dataLen + 1];
                            double[] pm10 = new double[dataLen + 1];

                            // 每个距离的值
                            for (int i = 0; i < dataLen + 1; i++) {
                                double lend = (double) i * radarSitePo.getRadius() / 1000000;
                                distance[i] = lend;
                                for (int j = 1; j < rowDatas.length; j++) {
                                    if (null != rowDatas[j]) {
                                        String[] distanceData = rowDatas[j].split(",");
                                        if (lend == ConvertUtil.toDouble(distanceData[0])) {
                                            // 每个点位的pm10,pm25值
                                            String[] data = distanceData[1].split("/");
                                            pm10[i] = ConvertUtil.toDouble(data[0]);
                                            pm2_5[i] = ConvertUtil.toDouble(data[1]);
                                            break;
                                        }
                                    }
                                }
                            }

                            byte[] beforePM25 = JSON.toJSONBytes(pm2_5);
                            byte[] beforePM10 = JSON.toJSONBytes(pm10);
                            byte[] afterPM25 = ZLibUtils.compress(beforePM25);
                            byte[] afterPM10 = ZLibUtils.compress(beforePM10);

                            singleDataPo.setDISTANCE(distance);
                            singleDataPo.setPM2_5(Base64Util.base64(afterPM25));
                            singleDataPo.setPM10(Base64Util.base64(afterPM10));
                            // 数据创建时间
                            singleDataPo.setCREATETIME(sdf.format(calendar.getTime()));
                            // 数据做处理（查缺补漏）

                            radarService.checkMissingTrapSingleData(singleDataPo);

                            log.info("数据处理完成:" + sdf.format(new Date()));
                            log.info("----------------------------------------------------------");

                            //删除队列中的消息
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                            // 单条数据推送到交换，供前端使用
                            JSONObject obj = new JSONObject();
                            obj.put("angle", singleDataPo.getANGLE());
                            obj.put("sncode", singleDataPo.getRADARCODE());
                            obj.put("monitorTime", singleDataPo.getMONITORTIME());
                            obj.put("pm2_5", pm2_5);
                            obj.put("pm10", pm10);
                            obj.put("distance", distance);
                            messageProducer.sendMessage("radarDataShow", "radar." + singleDataPo.getRADARCODE(), obj.toJSONString());
                            return;
                        }
                    }
                }
            }
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } catch (Exception e) {
            e.printStackTrace();
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }

}
